package com.baosight.xdata.zyl;

import com.baosight.xdata.redis.ConsumerInfo;
import com.baosight.xdata.redis.LettuceUtil;
import com.baosight.xdata.redis.StreamConsumerInfo;
import io.lettuce.core.*;
import io.lettuce.core.models.stream.PendingMessage;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

public class TssRedisStreamConsumer {
    public static String STREAM_1 = "zyl-stream:1";
    public static String STREAM_2 = "zyl-stream:2";

    public static String GROUP_1 = "zyl-group1";
    public static String GROUP_2 = "zyl-group2";
    private int batch = 50;

    private void initConsumer() {
        Set<RedisURI> uris = new HashSet<>();
        uris.add(RedisURI.builder().withHost("10.25.10.102").withPort(6378).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.102").withPort(6377).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.102").withPort(6379).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.104").withPort(6378).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.104").withPort(6377).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.104").withPort(6379).withTimeout(Duration.ofSeconds(30)).build());
        LettuceUtil.init(uris);
    }

    public void consumerStreamViaGroup() {
        MessagesReleaseManager messagesReleaseManager = new MessagesReleaseManager("TSS-redis-stream-monitor");
        messagesReleaseManager.start();
        try {
            XReadArgs xReadArgs = new XReadArgs();
            xReadArgs.block(Duration.ofSeconds(10));
            xReadArgs.count(batch);
            List<StreamMessage<String, String>> list1;
            List<StreamMessage<String, String>> list2;
            List<StreamMessage<String, String>> list3;
            List<StreamMessage<String, String>> list4;
            while (true) {
                try{
                    list1 = LettuceUtil.xreadGroupWithArgs(STREAM_1, GROUP_1, "zyl-consumer1", xReadArgs);
                    handlerMesasges(list1);
                    LettuceUtil.xdel(STREAM_1, list1);
                    LettuceUtil.xack(STREAM_1, list1, GROUP_1);
                    list3 = LettuceUtil.xreadGroupWithArgs(STREAM_1, GROUP_1, "zyl-consumer3", xReadArgs);
                    handlerMesasges(list3);
                    LettuceUtil.xdel(STREAM_1, list3);
                    LettuceUtil.xack(STREAM_1, list3, GROUP_1);
                }catch (Exception e){
                    System.out.println(String.format("stream:[%s] consumer redis exception, details : [%s] [%s]",
                            STREAM_1, e.getCause(), e.getMessage()));
                }

                try{
                    list2 = LettuceUtil.xreadGroupWithArgs(STREAM_2, GROUP_2, "zyl-consumer2", xReadArgs);
                    handlerMesasges(list2);
                    LettuceUtil.xdel(STREAM_2, list2);
                    LettuceUtil.xack(STREAM_2, list2, GROUP_2);
                    list4 = LettuceUtil.xreadGroupWithArgs(STREAM_2, GROUP_2, "zyl-consumer4", xReadArgs);
                    handlerMesasges(list4);
                    LettuceUtil.xdel(STREAM_2, list4);
                    LettuceUtil.xack(STREAM_2, list4, GROUP_2);
                }catch (Exception e){
                    System.out.println(String.format("stream:[%s] consumer redis exception, details : [%s] [%s]",
                            STREAM_2, e.getCause(), e.getMessage()));
                }
            }
        } catch (Exception e) {
            System.out.println(String.format("consumer redis exception, details : [%s] [%s]", e.getCause(), e.getMessage()));
        }
    }


    private void handlerMesasges(List<StreamMessage<String, String>> messageList) {
        if (!messageList.isEmpty()) {
            StringBuilder stringBuilder = new StringBuilder("=============");
            messageList.stream().forEach(x -> stringBuilder.append(x.toString()).append("\n"));
            System.out.println(stringBuilder.append("=============\n\n").toString());
        }
    }

    public static void main(String[] args) {
        TssRedisStreamConsumer tssRedisStreamConsumer = new TssRedisStreamConsumer();
        tssRedisStreamConsumer.initConsumer();
        tssRedisStreamConsumer.consumerStreamViaGroup();
    }

    private static class MessagesReleaseManager extends Thread{
        private int batch = 5;

        public MessagesReleaseManager(String name) {
            super(name);
        }

        @Override
        public void run() {
            System.out.println("=== MessagesReleaseManager start ===");
            while (true){
                try {
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    XReadArgs xReadArgs = new XReadArgs();
                    xReadArgs.block(Duration.ofSeconds(5));
                    xReadArgs.count(batch);
                    List<StreamConsumerInfo> streamConsumerInfoList = new ArrayList<>();
                    streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_1, GROUP_1, null));
                    streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_2, GROUP_2, null));
                    for (StreamConsumerInfo consumer : streamConsumerInfoList) {
                        System.out.println(String.format("========== %s ==========", consumer.toString()));
                        consumerPendingViaGroup(consumer);
                    }
                }catch (Exception e){
                 System.out.println(String.format("monitor PEL and release messages exception :[%s]", e.getMessage()));
                }
            }
        }

        private void consumerPendingViaGroup(StreamConsumerInfo streamConsumerInfo) {
            XReadArgs xReadArgs = new XReadArgs();
            xReadArgs.block(Duration.ofSeconds(5));
            xReadArgs.count(batch);
            String lowerBound = "-";
            String upperBound = "+";
            List<ConsumerInfo> consumerInfo = getConsumerInfo(streamConsumerInfo);
            Range<String> range = Range.from(Range.Boundary.including(lowerBound), Range.Boundary.including(upperBound));
            for (ConsumerInfo entry : consumerInfo) {
                System.out.println(String.format("\n\nstream:[%s] group:[%s] consumer:[%s]", streamConsumerInfo.getStream(), streamConsumerInfo.getGroup(), streamConsumerInfo.getGroup()));
                if (entry.getPending() <= 0) {
                    System.out.println("pending size [0]");
                } else {
                    for (long i = entry.getPending(); i > 0; ) {
                        List<PendingMessage> pendingMessageList = LettuceUtil.xpendingGroup(streamConsumerInfo.getStream(),
                                streamConsumerInfo.getGroup(), entry.getName(), range
                                , Limit.from(batch));
                        System.out.println(String.format("\n\nstream:[%s] group:[%s] consumer:[%s] pending size [%d]",
                                streamConsumerInfo.getStream(),
                                streamConsumerInfo.getGroup(),
                                streamConsumerInfo.getGroup(),
                                pendingMessageList.size()
                        ));
                        pendingMessageList.forEach(x -> System.out.println(x.toString()));
                        List<String> collect = pendingMessageList.stream().map(PendingMessage::getId).collect(Collectors.toList());
                        LettuceUtil.xdelWithBatch(streamConsumerInfo.getStream(), collect);
                        LettuceUtil.xackWithBatch(streamConsumerInfo.getStream(), streamConsumerInfo.getGroup(), collect);
                        i = i - (pendingMessageList.size() > 0 ?  pendingMessageList.size() : batch);
                    }
                }
            }
        }

        private List<ConsumerInfo> getConsumerInfo(StreamConsumerInfo streamConsumerInfo) {
            List<ConsumerInfo> ConsumerInfoList = new ArrayList<>();
            List<Object> consumerInfo = LettuceUtil.getConsumerInfo(streamConsumerInfo.getStream(), streamConsumerInfo.getGroup());
            for (Object entry : consumerInfo) {
                ArrayList<Object> list = (ArrayList<Object>) entry;
                ConsumerInfoList.add(ConsumerInfo.Builder(list));
            }
            return ConsumerInfoList;
        }
    }
}
